package com.zhangxin.aiInterview.manager.wsserver;

import cn.hutool.core.util.ObjUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.zhangxin.aiInterview.common.ErrorCode;
import com.zhangxin.aiInterview.constant.FileConstant;
import com.zhangxin.aiInterview.exception.BusinessException;
import com.zhangxin.aiInterview.manager.cos.CosManager;
import com.zhangxin.aiInterview.manager.work_flow.ai_hr_interviewer.AiHrInterviewerAgentFacade;
import com.zhangxin.aiInterview.manager.work_flow.ai_tec_interviewer.AiTecInterviewerAgentFacade;
import com.zhangxin.aiInterview.manager.asr.ASRWs;
import com.zhangxin.aiInterview.manager.asr.ASRWsManager;
import com.zhangxin.aiInterview.manager.python_api.PythonApi;
import com.zhangxin.aiInterview.manager.redis.audio_emotion.SetAudioEmotion;
import com.zhangxin.aiInterview.manager.redis.stress_detector.StressDetectionProducer;
import com.zhangxin.aiInterview.manager.wsserver.enums.WsMessageTypeEnum;
import com.zhangxin.aiInterview.manager.wsserver.model.WsResponseMessage;
import com.zhangxin.aiInterview.manager.disruptor.WsEventProducer;
import com.zhangxin.aiInterview.model.entity.HrInterview;
import com.zhangxin.aiInterview.model.entity.ResumeContent;
import com.zhangxin.aiInterview.model.entity.AiInterview;
import com.zhangxin.aiInterview.model.enums.InterviewTypeEnum;
import com.zhangxin.aiInterview.service.AiInterviewService;
import com.zhangxin.aiInterview.service.HrInterviewService;
import com.zhangxin.aiInterview.service.ResumeContentService;
import com.zhangxin.aiInterview.utils.AudioEmotionsSequenceManager;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomStringUtils;
import org.java_websocket.client.WebSocketClient;
import org.jetbrains.annotations.NotNull;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.*;

import javax.annotation.Resource;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

/**
 * WebSocket消息处理器处理所有消息
 */
@Component
@Slf4j
public class WsMessageHandler implements WebSocketHandler {
    //单例
    //管理用户音频的Map集合（key为ws连接， value为这个ws连接的音频缓存）
    public static final Map<WebSocketSession, ByteArrayOutputStream> audioBufferMap = new ConcurrentHashMap<>();

    private static boolean isStart = false;
    @Resource
    private AiTecInterviewerAgentFacade aiTecInterviewerAgentFacade;
    @Resource
    private AiHrInterviewerAgentFacade hrInterviewerAgentFacade;
    @Resource
    private WsEventProducer wsEventProducer;
    @Resource
    private ResumeContentService resumeContentService;
    @Resource
    private PythonApi pythonApi;
    @Resource
    private StressDetectionProducer stressDetectionProducer;
    @Resource
    private AiInterviewService aiInterviewService;
    @Resource
    private HrInterviewService hrInterviewService;
    @Resource
    private SetAudioEmotion setAudioEmotion;
    @Resource
    private AudioEmotionsSequenceManager audioEmotionsSequenceManager;
    @Resource
    private CosManager cosManager;
    public static final String AUDIO_PREFIX = "ai_interview/user_audio";

    /**
     * 处理 WebSocket 消息
     *
     * @param session 当前发送来消息的ws连接会话
     * @param message
     * @throws Exception
     */
    @Override
    public void handleMessage(@NotNull WebSocketSession session, WebSocketMessage<?> message) throws Exception {
        //获取当前ws连接的临时存储数组
        ByteArrayOutputStream audioBuffer = audioBufferMap.computeIfAbsent(session, s -> new ByteArrayOutputStream());
        if ("start".equals(message.getPayload())) isStart = true;
        if (!isStart) return;

        //临时存储WebM音频数据
        if (message.getPayload() instanceof ByteBuffer) {
            // 将接收到的数据拼接到临时数组中
            ByteBuffer byteBuffer = (ByteBuffer) message.getPayload();
            byte[] bytes = byteBuffer.array();
            audioBuffer.write(bytes);
        }
        // 用户录音结束，可以开始转换
        if (isEndOfStream(message)) {
            isStart = false;

            // 将临时数组中存储的音频数据保存到临时文件
            long timestamp = System.currentTimeMillis();
            String uuid = RandomStringUtils.randomAlphanumeric(8);
            File tempWebMFile = new File(uuid + "-" + timestamp + ".webm");
            try (FileOutputStream fos = new FileOutputStream(tempWebMFile)) {
                fos.write(audioBuffer.toByteArray());
            }
            //获取用户id和面试类型
            Long userId = (Long) session.getAttributes().get("userId");
            Integer interviewType = (Integer) session.getAttributes().get("interviewType");
            //将音频文件存储到COS
            String filepath = String.format("/%s/%s/%s", AUDIO_PREFIX, userId, tempWebMFile.getName());
            cosManager.putObject(filepath, tempWebMFile);
            String audioUrl = FileConstant.COS_HOST + filepath;
            System.out.println("音频文件上传Cos成功");

            //清理临时数组中的音频数据
            audioBuffer.reset();

//            //获取音频绝对路径
//            String audioAbsolutePath = tempWebMFile.getAbsolutePath();

            // 1. 识别音频情感
            String emotionResponse = pythonApi.analyzeEmotion(audioUrl);
            JSONObject emotionJsonObject;
            try {
                emotionJsonObject = JSONUtil.parseObj(emotionResponse);
            } catch (Exception e) {
                System.out.println("情感分析返回的JSON格式错误: " + emotionResponse);
                // 创建一个默认的JSON对象，模拟接口正常返回但无情感识别的情况
                emotionJsonObject = JSONUtil.createObj()
                        .set("emotion", null)
                        .set("emotion_sequence", null)
                        .set("detail", "JSON解析失败: " + e.getMessage());
            }

            // 获取情感结果，使用null安全的方式
            Object emotionObj = emotionJsonObject.get("emotion");
            String emotionSequence = emotionJsonObject.getStr("emotion_sequence", "");
            System.out.println("情感序列：" + emotionSequence);
            System.out.println("情感结果：" + emotionObj);

            // 判断是否为空
            if (ObjUtil.isEmpty(emotionObj)) {
                // 存储默认情感数据
                setAudioEmotion.storeEmotion(userId, "Calm (冷静)");
                if (session.isOpen()) {
                    WsResponseMessage response = new WsResponseMessage(
                            WsMessageTypeEnum.AUDIO_EMOTION.getValue(),
                            "{\"emotion\": \"Calm (冷静)\"}"
                    );
                    // 发送结果
                    session.sendMessage(new TextMessage(JSONUtil.toJsonStr(response)));
                }
            } else {
                // 存储情感数据
                setAudioEmotion.storeEmotion(userId, emotionObj.toString());
                // 发送情感结果
                if (session.isOpen()) {
                    // 创建一个新的JSON对象只包含emotion字段，避免发送多余信息
                    String emotionJson = JSONUtil.createObj().set("emotion", emotionObj).toString();
                    WsResponseMessage response = new WsResponseMessage(
                            WsMessageTypeEnum.AUDIO_EMOTION.getValue(),
                            emotionJson
                    );
                    session.sendMessage(new TextMessage(JSONUtil.toJsonStr(response)));
                }

                // 存储音频序列情感，确保不为null
                if (StrUtil.isNotEmpty(emotionSequence)) {
                    audioEmotionsSequenceManager.addEmotionSequence(userId, emotionSequence);
                }
            }

            //2. 检测压力值
            stressDetectionProducer.sendDetectionTask(userId, audioUrl, interviewType);

            //3. 异步语音识别
            //建立ws连接
            WebSocketClient webSocketClient = ASRWsManager.getConnection(userId);
            //兜底：查看asr连接是否断开，如果断开就重新建立连接
            if (webSocketClient == null || !webSocketClient.isOpen()) {
                ASRWs.init(userId);
                webSocketClient = ASRWsManager.getConnection(userId);
            }
            JSONObject jsonObject = new JSONObject();
            jsonObject.set("path", audioUrl);
            webSocketClient.send(jsonObject.toString());
            //删除临时音频文件
            tempWebMFile.delete();
        }
    }

    /**
     * 判断是否是音频数据的结束（例如判断是否已传输完）
     *
     * @param message
     * @return
     */
    private boolean isEndOfStream(WebSocketMessage<?> message) {
        return "stop".equals(message.getPayload());
    }

    /**
     * 连接建立成功回调
     *
     * @param session 当前ws连接会话
     * @throws Exception
     */
    @Override
    public void afterConnectionEstablished(WebSocketSession session) {
        System.out.println("实时语音对话，客户端和服务端WebSocket连接成功");
        //管理用户客户端服务端ws连接;
        Long userId = (Long) session.getAttributes().get("userId");
        //获取面试类型
        Integer interviewType = (Integer) session.getAttributes().get("interviewType");
        WsManager.addOrUpdateConnection(userId, session);

        //建立ASR 语音识别ws连接
        ASRWs.init(userId);

        //异步执行
        //匿名内部类实际上是一个对象，它会持有外部类实例的引用和外部方法参数的引用，将它作为成员变量，便于后续访问
        CompletableFuture.runAsync(() -> {
            try {
                //获取用户面试信息
                List<String> list = generateAiInterviewInfoWithUser(userId, interviewType);
                String aiInterviewInfoWithUser = list.get(0);
                String interviewerGender = list.get(1);
                //调用AI面试官智能体生成面试题
                String aiQuestion;
                if (Objects.equals(InterviewTypeEnum.TEC_INTERVIEW.getValue(), interviewType)) {
                    System.out.println("开始异步技术模拟面试");
                    //技术面试
                    aiQuestion = aiTecInterviewerAgentFacade.sendAndGetReply(userId, aiInterviewInfoWithUser);
                } else {
                    System.out.println("开始异步hr模拟面试");
                    //hr面试
                    aiQuestion = hrInterviewerAgentFacade.sendAndGetReply(userId, aiInterviewInfoWithUser);
                }

                //异步 文本转语音(disruptor)
//                wsEventProducer.publishEvent(userId, aiQuestion, interviewerGender);
                //返回音频对应的文本信息给客户端
                WsResponseMessage wsResponseMessage = new WsResponseMessage(WsMessageTypeEnum.AUDIO_TEXT.getValue(), aiQuestion);
                session.sendMessage(new TextMessage(JSONUtil.toJsonStr(wsResponseMessage)));
            } catch (IOException e) {
                throw new BusinessException(ErrorCode.OPERATION_ERROR, "返回音频文本信息异常");
            }
        }).exceptionally(ex -> {
            log.error("异步Ai模拟面试生成面试问题异常：", ex);
            return null;
        });
    }

    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) {
    }

    /**
     * ws连接关闭事件回调
     *
     * @param session     当前ws连接
     * @param closeStatus 错误信息
     */
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) {
        System.out.println("实时语音对话，客户端和服务端WebSocket连接关闭");
        Long userId = (Long) session.getAttributes().get("userId");
        //清理资源
        WsManager.removeConnection(userId);
        ASRWsManager.removeConnection(userId);
        audioBufferMap.remove(session);
        System.out.println("清理用户客户端和服务端ws连接");
        System.out.println("清理用户ASR语音识别ws连接");
        System.out.println("清理用户临时存储音频数组");
        System.out.println("清理用户session锁对象");
    }

    @Override
    public boolean supportsPartialMessages() {
        return false;
    }

    /**
     * 生成本轮ai模式面试的信息，ai面试官根据信息来提问
     *
     * @param
     * @param userId 用户id
     * return 返回一个集合，集合中包含面试信息和面试官性别
     */
    private List<String> generateAiInterviewInfoWithUser(Long userId, Integer interviewType) {
        String company;
        String position;
        Long resumeId;
        String language;
        String interviewerGender;
        String interviewerType;

        if (Objects.equals(interviewType, InterviewTypeEnum.TEC_INTERVIEW.getValue())) {
            //获取用户技术面试信息
            AiInterview aiInterview = aiInterviewService.lambdaQuery()
                    .eq(AiInterview::getUserId, userId)
                    .orderBy(true, false, AiInterview::getId)
                    .last("LIMIT 1")
                    .one();

            company = aiInterview.getCompany();
            position = aiInterview.getPosition();
            resumeId = aiInterview.getResumeId();
            language = aiInterview.getLanguage();
            interviewerGender = aiInterview.getInterviewer();
            interviewerType = aiInterview.getInterviewerType();
        } else {
            //获取用户hr面试信息
            HrInterview hrInterview = hrInterviewService.lambdaQuery()
                    .eq(HrInterview::getUserId, userId)
                    .orderBy(true, false, HrInterview::getId)
                    .last("LIMIT 1")
                    .one();

            company = hrInterview.getCompany();
            position = hrInterview.getPosition();
            resumeId = hrInterview.getResumeId();
            language = hrInterview.getLanguage();
            interviewerGender = hrInterview.getInterviewer();
            interviewerType = hrInterview.getInterviewerType();
        }

        //获取用户简历内容
        ResumeContent resumeContent = resumeContentService.lambdaQuery()
                .eq(ResumeContent::getResumeId, resumeId)
                .one();
        //获取每个部分的内容
        String education = resumeContent.getEducation();
        String skill = resumeContent.getSkill();
        String internship = resumeContent.getInternship();
        String projectExperience = resumeContent.getProjectExperience();
        String workExperience = resumeContent.getWorkExperience();
        String schoolExperience = resumeContent.getSchoolExperience();
        String certificate = resumeContent.getCertificate();
        String evaluation = resumeContent.getEvaluation();

        String json = "{"
                + "\"resume\":\""
                + "【教育经历】:" + education + "\\n"
                + "【专业技能】:" + skill + "\\n"
                + "【实习经历】:" + internship + "\\n"
                + "【项目经历】:" + projectExperience + "\\n"
                + "【校园经历】:" + schoolExperience + "\\n"
                + "【工作经历】:" + workExperience + "\\n"
                + "【获奖证书】:" + certificate + "\\n"
                + "【自我评价】:" + evaluation
                + "\"," + "\\n"
                + "\"company\":\"" + company + "\"," + "\\n"
                + "\"position\":\"" + position + "\"," + "\\n"
                + "\"interviewerType\":\"" + interviewerType + "\"," + "\\n"
                + "\"questionNumber\": 4" + "\\n" //最少面试轮数
                + "\"language\":" + language + "\\n"
                + "}";
        log.info(json);
        ArrayList<String> list = new ArrayList<>();
        list.add(json);
        list.add(interviewerGender);
        return list;
    }
}